package com.stress.testing.plugin.apache.kafka;

import com.stress.testing.common.matcher.ClassDescription;
import com.stress.testing.common.matcher.ClassMatcher;
import com.stress.testing.common.matcher.MethodDescription;
import com.stress.testing.common.matcher.MethodMatcher;
import com.stress.testing.common.plugin.AbstractPlugin;

import java.util.ArrayList;
import java.util.List;

/**
 * pull模式消费，能做的事情不多
 */
public class KafkaConsumerPlugin extends AbstractPlugin {
    @Override
    public ClassMatcher getClassMatcher() {
        return new ClassMatcher() {
            @Override
            public boolean match(ClassDescription classDescription) {
                return "org.apache.kafka.clients.consumer.KafkaConsumer".equals(classDescription.getName());
            }


            // pause\resume貌似拦截也没啥意义
            @Override
            public List<MethodMatcher> getMethodMatchers() {
                List<MethodMatcher> methodMatchers = new ArrayList<>();

                methodMatchers.add(new MethodMatcher() {
                    @Override
                    public boolean match(MethodDescription methodDescription) {
                        return "poll".equals(methodDescription.getName());
                    }

                    @Override
                    public String getInterceptor() {
                        return "com.stress.testing.plugin.apache.kafka.interceptor.KafkaConsumerPollInterceptor";
                    }
                });

                methodMatchers.add(new MethodMatcher() {
                    @Override
                    public boolean match(MethodDescription methodDescription) {
                        return "subscribe".equals(methodDescription.getName());
                    }

                    @Override
                    public String getInterceptor() {
                        return "com.stress.testing.plugin.apache.kafka.interceptor.KafkaConsumerSubscribeInterceptor";
                    }
                });

                methodMatchers.add(new MethodMatcher() {
                    @Override
                    public boolean match(MethodDescription methodDescription) {
                        return "unsubscribe".equals(methodDescription.getName());
                    }

                    @Override
                    public String getInterceptor() {
                        return "com.stress.testing.plugin.apache.kafka.interceptor.KafkaConsumerUnsubscribeInterceptor";
                    }
                });

                methodMatchers.add(new MethodMatcher() {
                    @Override
                    public boolean match(MethodDescription methodDescription) {
                        return "close".equals(methodDescription.getName());
                    }

                    @Override
                    public String getInterceptor() {
                        return "com.stress.testing.plugin.apache.kafka.interceptor.KafkaConsumerCloseInterceptor";
                    }
                });

                return methodMatchers;
            }
        };
    }
}
